package redis;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.redis.bolt.RedisStoreBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisStoreMapper;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;

public class PersistentWordCount {
    public static final String HASH_KEY_NAME = "WordCount";

    public static void main(String[] args) throws Exception {
        // 创建配置对象
        JedisPoolConfig liuConfig = new JedisPoolConfig.Builder().setHost("liu").setPort(6379).build();

        // 基本组件实例化
        WordSpout wordSpout = new WordSpout();
        WordCountBolt wordCountBolt = new WordCountBolt();
        WordCountStoreMapper wordCountStoreMapper = new WordCountStoreMapper();
        RedisStoreBolt redisStoreBolt = new RedisStoreBolt(liuConfig, wordCountStoreMapper);

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("wordSpout", wordSpout);
        builder.setBolt("wordCountBolt", wordCountBolt)
                .fieldsGrouping("wordSpout", new Fields("word"));
        builder.setBolt("redisStoreBolt", redisStoreBolt)
                .shuffleGrouping("wordCountBolt");

        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("store-topology", config, builder.createTopology());

    }

    /**
     * 定义 Tuple 与 Redis 中数据的映射关系
     */
    private static class WordCountStoreMapper implements RedisStoreMapper {
        private final RedisDataTypeDescription description;

        public WordCountStoreMapper() {
            this.description = new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH, HASH_KEY_NAME);
        }

        @Override
        public RedisDataTypeDescription getDataTypeDescription() {
            return description;
        }

        /**
         * 根据这两个方法获取上游Tuple发来的字段，并作为Hash中的Key和Value
         */
        @Override
        public String getKeyFromTuple(ITuple iTuple) {
            return iTuple.getStringByField("word");
        }

        @Override
        public String getValueFromTuple(ITuple iTuple) {
            return iTuple.getStringByField("count");
        }
    }
}
